package org.apache.spark.examples;

// 引入必要的Java库和Scala类型
import java.util.ArrayList;
import java.util.List;
import java.util.regex.Pattern;

import scala.Tuple2;

import com.google.common.collect.Iterables;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.sql.SparkSession;

/**
 * 使用Spark计算URL的PageRank。输入文件应为以下格式：
 * URL         相邻URL
 * URL         相邻URL
 * ...
 * 其中URL和相邻URL之间由空格分隔。
 *
 * 这是一个用于学习如何使用Spark的示例实现。对于更常规的使用，请参考org.apache.spark.graphx.lib.PageRank
 *
 * 示例用法：
 * <pre>
 * bin/run-example JavaPageRank data/mllib/pagerank_data.txt 10
 * </pre>
 */
public final class JavaPageRankZH {
  // 用于分割URL和相邻URL的正则表达式
  private static final Pattern SPACES = Pattern.compile("\\s+");

  // 显示警告信息，说明这是一个简单的PageRank实现
  static void showWarning() {
    String warning = "WARN: This is a naive implementation of PageRank " +
            "and is given as an example! \n" +
            "Please use the PageRank implementation found in " +
            "org.apache.spark.graphx.lib.PageRank for more conventional use.";
    System.err.println(warning);
  }

  // 定义一个用于累加Double值的Function2
  private static class Sum implements Function2<Double, Double, Double> {
    @Override
    public Double call(Double a, Double b) {
      return a + b;
    }
  }

  // 主函数，程序的入口
  public static void main(String[] args) throws Exception {
    // 检查命令行参数
    if (args.length < 2) {
      System.err.println("Usage: JavaPageRank <file> <number_of_iterations>");
      System.exit(1);
    }

    // 显示警告信息
    showWarning();

    // 初始化SparkSession
    SparkSession spark = SparkSession
      .builder()
      .appName("JavaPageRank")
      .getOrCreate();

    // 加载输入文件，格式如上面所述

    JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();

    // 解析输入数据，建立URL到其相邻URL的映射，并缓存结果
    JavaPairRDD<String, Iterable<String>> links = lines.mapToPair(s -> {
      String[] parts = SPACES.split(s);
      return new Tuple2<>(parts[0], parts[1]);
    }).distinct().groupByKey().cache();

    // 初始化所有URL的排名为1.0
    JavaPairRDD<String, Double> ranks = links.mapValues(rs -> 1.0);

    // 执行指定次数的PageRank迭代
    for (int current = 0; current < Integer.parseInt(args[1]); current++) {
      // 计算每个URL对其它URL排名的贡献
      JavaPairRDD<String, Double> contribs = links.join(ranks).values()
        .flatMapToPair(s -> {
          int urlCount = Iterables.size(s._1());
          List<Tuple2<String, Double>> results = new ArrayList<>();
          for (String n : s._1) {
            results.add(new Tuple2<>(n, s._2() / urlCount));
          }
          return results.iterator();
        });

      // 根据贡献更新每个URL的排名
      ranks = contribs.reduceByKey(new Sum()).mapValues(sum -> 0.15 + sum * 0.85);
    }

    // 收集并打印所有URL的排名
    List<Tuple2<String, Double>> output = ranks.collect();
    for (Tuple2<?,?> tuple : output) {
      System.out.println(tuple._1() + " has rank: " + tuple._2() + ".");
    }

    // 停止SparkSession
    spark.stop();
  }
}